package com.baishancloud.log.live.tengine

import com.baishancloud.log.common.util.TimeUtil
import com.baishancloud.log.format.java.util.IPLocator
import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import java.text.SimpleDateFormat
import java.util.{Locale, Objects, Timer, TimerTask}


class TengineRecordParse(parameterTool: ParameterTool) extends ProcessFunction[Array[String], TengineLog] {

  val LOGGER: Logger = LoggerFactory.getLogger(classOf[TengineRecordParse])
  //只取日志时间的秒为指定字符串的日志
  val allowSeconds: Array[String] = Array("00", "01", "02", "03")
  //去掉域名中的其他特殊字符
  val regex: String = "[^0-9a-zA-Z-_.]"

  var allowOldest: Long = parameterTool.getInt(allowOldestHour, 6) * 3600 * 1000

  override def open(parameters: Configuration): Unit = {
    new Timer("allowOldestUpdate", true).scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = allowOldest = parameterTool.getInt(allowOldestHour, 6) * 3600 * 1000
    }, 60000, 60000)
  }

  override def processElement(value: Array[String], ctx: ProcessFunction[Array[String], TengineLog]#Context, out: Collector[TengineLog]): Unit = {
    //ip
    val ip: String = value(1)
    val ipInfo = IPLocator.lookup(ip)

    var isp = ipInfo.getIsp.split("/")(0)
    if (isp.equals("other")) isp = "其他"
    if (isp.contains("广电")) {
      isp = "广电"
    } else if (isp.contains("华数")) {
      isp = "华数"
    }

    val country = if (ipInfo.getCountry.equals("other")) "其他" else ipInfo.getCountry
    var province = if (ipInfo.getProvince.equals("other")) "其他" else ipInfo.getProvince
    if ((country.equals("中国") || country.startsWith("CHINA")) && province.equals("其他")) province = "中国其他地区"

    //domain
    var domain: String = value(2)
    domain = StringUtils.replacePattern(domain, regex, "")

    //日志时间
    val logTime: String = s"${value(4)} ${value(5)}"
    val logMillisecond: Long = timeStrToMillisecond(logTime)
    //直接将毫秒值转化为对应字符串，用于下面的时间取样过滤
    var timeAt1Min: String = TimeUtil.longToString(logMillisecond)


    //过滤
    //状态码，只要2xx的。
    val statusCode: Int = value(9).toInt
    val v18: Array[String] = value(18).split("@")
    val v20: String = value(20)
    val b1: Boolean = statusCode >= 200 && statusCode < 300
    val b2: Boolean = (v18.length >= 23 && Objects.equals(v18(22), "-")) || v18.length < 23
    val b3: Boolean = Objects.equals(v20, "\"-\"")
    //只取每分钟内前4秒的日志
    val b4: Boolean = allowSeconds.contains(timeAt1Min.substring(timeAt1Min.lastIndexOf(":") + 1))
    //去掉迟到6小时数据、去掉未来时间数据
    //    val b5: Boolean = logMillisecond > allowOldest && logMillisecond <= current
    val b5: Boolean = logMillisecond > allowOldest

    if (b1 && b2 && b3 && b4 && b5) {
      //重新将时间归到需要的时间点
      timeAt1Min = TimeUtil.timeAtStringS(timeAt1Min, 60000)
      out.collect(TengineLog(timeAt1Min, domain, country, province, isp, 2))
    }
  }


  /**
   * 将日志中的时间转化为毫秒值
   */
  def timeStrToMillisecond(timeStr: String): Long = {
    val sdf = new SimpleDateFormat("[dd/MMM/yyyy:HH:mm:ss Z]", Locale.ENGLISH)
    sdf.parse(timeStr).getTime
  }
}
